    with send_bill as (
    select t.* from (
            select
                t.scan_site,
                t.scan_site_code,
                t.waybill_no,
                t.scan_time,
                fj.input_site last_site_name,
                fj.scan_site_code as last_site_code,
                t.next_station,
                t.next_station_code as next_site_code,
                (case
                    when b.is_distribution_center = 2 then 1
                    when a.is_distribution_center = 1
                    and b.is_distribution_center = 1 then 2
                    when a.is_distribution_center = 2
                    and b.is_distribution_center = 1 then 3
                    else 0
                end) as type,
                row_number() over(partition by t.scan_site_code,t.waybill_no order by t.scan_time desc,fj.scan_time desc) as rid
            from
                jms_dwd.dwd_tab_barscan_centersend_base_dt t
            left join (select * from jms_dwd.dwd_tab_barscan_send_base_dt t where t.dt>='{{ execution_date | date_add(-15) | cst_ds }}' and t.dt<='{{ execution_date  | cst_ds }}'
                                                and to_date(t.scan_time)>='{{ execution_date | date_add(-7) | cst_ds }}' and to_date(t.scan_time) <='{{ execution_date | cst_ds }}'
            ) fj  on
                fj.waybill_no = t.waybill_no
                and fj.scan_time < t.scan_time
                and fj.next_station_code = t.scan_site_code

            left join jms_dim.dim_network_whole_massage a  on
                a.code = fj.scan_site_code
            left join jms_dim.dim_network_whole_massage b on
                b.code = t.next_station_code

            where
                t.dt>='{{ execution_date | date_add(-15) | cst_ds }}' and  t.dt<='{{ execution_date | cst_ds }}'
                 and t.waybill_no not like 'B%'
                 and to_date(t.scan_time) ='{{ execution_date | cst_ds }}'
            ) t where rid=1
    ),

    base_bill as (
    select
      t.waybill_no
     ,max(t.package_receipt_weight)  as package_receipt_weight
     ,max(t.package_collect_weight) as package_collect_weight
     ,sum(case when send_center_station_number=1 then send_station_max_weight else 0 end ) as first_center_weight
     ,sum(case when send_entrepot_station_number=1 then send_station_max_weight else 0 end ) as first_site_weight
     ,t.ordersource_code as waybill_source_code
     ,t.ordersource_name as waybill_source_name
    from jms_dwd.dwd_wide_unsign_trace_waybill_step_dt t
    where t.dt >= '{{ execution_date | date_add(-15) | cst_ds }}' and t.dt <= '{{ execution_date | cst_ds }}'
    group by t.waybill_no  ,t.ordersource_code ,t.ordersource_name

    )

    insert overwrite table jms_dm.dm_transfer_bill_count_detail_dt partition (dt)
    --转运货量明细表
    select
      date_format(t.scan_time,'yyyy-MM-dd') as scan_time_ff
     ,t.waybill_no --运单号
     ,t.scan_site_code as site_code--本站点code
     ,t.scan_site as site_name--本站点name
     ,null as site_type--本站点类型
     ,b.manage_code as regional_id  --大区id
     ,b.manage_name as regional_desc  --大区
     ,b.agent_code --代理区
     ,b.agent_name --代理区
     ,t.next_site_code as next_site_code --下一站code
     ,t.next_station as next_site_name--下一站name
     ,null as next_site_type--下一站点类型，网点表
     ,null as  next_site_type2--下一站点类型，宽表
     ,t1.package_receipt_weight as  receipt_weight--收件重量，单位千克
     ,t1.package_collect_weight as collect_weight--入仓重量，单位千克
     ,t1.first_center_weight--首中心重量
     ,t1.first_site_weight--集散重量
     ,t.scan_time as scan_time--扫描时间
     ,t.last_site_code  as up_site_code--上一站code
     ,t.last_site_name  as up_site--上一站name
     ,null as up_site_type--上一站类型
     ,t.type as site_weight--本站重量
    --  ,coalesce(t.first_center_weight,t.first_site_weight,t.package_collect_weight,t.package_receipt_weight) as res_weight
    ,case when t1.first_center_weight>0 then t1.first_center_weight
          when t1.first_site_weight>0 then t1.first_site_weight
          when t1.package_collect_weight>0 then t1.package_collect_weight
          when t1.package_receipt_weight>0 then t1.package_receipt_weight else 0 end res_weight
     ,t1.waybill_source_code
     ,t1.waybill_source_name
     ,date_format(scan_time,'yyyy-MM-dd') as dt
    from send_bill t
    left join base_bill t1 on t.waybill_no=t1.waybill_no
    left join jms_dim.dim_network_whole_massage b on  b.code = t.scan_site_code
    where date_format(t.scan_time,'yyyy-MM-dd') = '{{ execution_date | cst_ds }}'
    
 distribute by pmod(hash(rand()), 140)
    ;